package com.yitiao.quickstart;

import com.yitiao.constants.Const;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.List;

/**
 * 消费者集群，监听生产者队列
 */
public class ConsumerModel {
    public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test_quick_consumer_name");
        consumer.setNamesrvAddr(Const.NAMESRV_ADDR_SINGLE);

        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        // 订阅topic
        consumer.subscribe(Const.TEST_QUICKSTART_TOPIC, "TagA"); // "*"
        // 集群
        consumer.setMessageModel(MessageModel.CLUSTERING);
//        consumer.setMessageModel(MessageModel.BROADCASTING);
        // 监听
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgList,
                                                            ConsumeConcurrentlyContext consumeConcurrentlyContext) {

                // TODO: 2022/5/6 批量发送还是单个发送
                MessageExt messageExt = msgList.get(0);
                try {

                    System.out.println(messageExt.getProperties().toString());
                    System.err.println(new String(messageExt.getBody(), RemotingHelper.DEFAULT_CHARSET));
//                    int a = 3/0;
                } catch (Exception e) {
                    // messagrDelayLevel  共重试15次
                    e.printStackTrace();
                    int reconsumeTimes = messageExt.getReconsumeTimes();
                    System.err.println("reconsumeTimes: " + reconsumeTimes);
                    if (reconsumeTimes == 3) {
                        System.out.println("------重试第三次，计入日志做补偿");
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();
        System.err.println("consumerModel start...");
    }
}
